package org.example.netty.model;

import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import lombok.extern.slf4j.Slf4j;
import org.example.netty.model.CustomMessage;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
@Slf4j
public class ExtendedChannelContext {
    /** 自定义定时任务线程池 */
    private static final ScheduledExecutorService SCHEDULER_POOL = new ScheduledThreadPoolExecutor(12, new ScheduledThreadFactory("extended-channel-context"));
    private static class ScheduledThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        ScheduledThreadFactory(String name) {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                    Thread.currentThread().getThreadGroup();
            namePrefix = (name==null||name==""?"pool-":name) +
                    poolNumber.getAndIncrement() +
                    "-thread-";
        }

        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                    namePrefix + threadNumber.getAndIncrement(),
                    0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }


    private final Map<Long/* 请求id */, CompletableFuture<CustomMessage>/* 异步返回future */> pendingFutures = new ConcurrentHashMap<>();
    private final Channel channel;
    private String clientId;
    public ExtendedChannelContext(Channel channel) {
        this.channel = channel;
    }
    public Channel getChannel(){
        return this.channel;
    }
    public void setClientId(String clientId){
        this.clientId=clientId;
    }
    public void close(){
        if (pendingFutures==null||pendingFutures.size()==0){
            return;
        }
        for (Map.Entry<Long, CompletableFuture<CustomMessage>> longCompletableFutureEntry : pendingFutures.entrySet()) {
            CompletableFuture<CustomMessage> value = longCompletableFutureEntry.getValue();
            value.completeExceptionally(new Exception("因为ExtendChannelContext被关闭了，所以下面的等待的future 我要主动都抛异常"));
        }
        pendingFutures.clear();
    }
    public void addRequest(long requestId,CompletableFuture<CustomMessage> future){
        pendingFutures.put(requestId,future);
    }
    public void completeRequest(long requestId,CustomMessage message){
        CompletableFuture<CustomMessage> future = this.removeRequest(requestId);
        if (future!=null&& !future.isDone()){
            future.complete(message);
        }
    }
    public CompletableFuture<CustomMessage> getPendingRequest(long requestId){
        return pendingFutures.get(requestId);
    }
    public CompletableFuture<CustomMessage> removeRequest(long requestId){
        return pendingFutures.remove(requestId);
    }

    /**
     * 同步阻塞发送请求包
     * @param requestId requestId全局唯一
     * @param message
     * @return 返回response中的消息
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public CustomMessage sendRequestAndWait(Long requestId,CustomMessage message) throws ExecutionException, InterruptedException {
        //todo 是否做校验
        CompletableFuture<CustomMessage> future = new CompletableFuture<>();
        addRequest(requestId,future);
        this.channel.writeAndFlush(message).addListener(f -> {
            if (!f.isSuccess()) {
                future.completeExceptionally(f.cause());
                removeRequest(requestId);
            }
        });
        return future.get();
    }
    /**
     * 同步阻塞发送请求包，有等待时间（推荐使用）
     * @param requestId requestId全局唯一
     * @param message
     * @return 返回response中的消息
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public CustomMessage sendRequestAndWait(Long requestId,CustomMessage message,long timeout, TimeUnit unit) throws ExecutionException, InterruptedException, TimeoutException {
        //todo 是否做校验
        CompletableFuture<CustomMessage> future = new CompletableFuture<>();
        addRequest(requestId,future);
        //todo 是否要讲 write之后的future放出去
        this.channel.writeAndFlush(message).addListener(f -> {
            if (!f.isSuccess()) {
                future.completeExceptionally(f.cause());
                removeRequest(requestId);
            }
        });
        return future.get(timeout,unit);
    }

    /**
     * 异步非阻塞 发送请求包
     * @param requestId 请求id
     * @param requestMsg 发送的消息
     * @param timeout 等待时间
     * @param unit 等待时间单位
     * @param action 实际处理 responseMessage和异常的回调函数
     * @return future
     */
    public ChannelFuture sendRequestAsync(Long requestId,CustomMessage requestMsg,long timeout, TimeUnit unit,BiConsumer<CustomMessage, ? super Throwable> action){
        //构建等待响应结果的 future
        CompletableFuture<CustomMessage> future = new CompletableFuture<>();
        //将requestId与future绑定
        addRequest(requestId,future);

        //发送消息
        ChannelFuture channelFuture = this.channel.writeAndFlush(requestMsg).addListener(f -> {
            if (!f.isSuccess()) {
                future.completeExceptionally(f.cause());
                removeRequest(requestId);
            }
        });
        //利用定时任务线程池，检查等待是否超时
        SCHEDULER_POOL.schedule(()->{
            if (future.isDone()){
                return;
            }
            future.completeExceptionally(new TimeoutException("请求超时requestId:"+requestId));
            this.removeRequest(requestId);
        },timeout,unit);
        //异步处理结果
        future.whenCompleteAsync((respMsg,ex)->{
            //接收到响应包，本次请求以完成，释放 requestId和future
            removeRequest(requestId);
            //处理自定义的回调函数
            action.accept(respMsg,ex);
/*            if (ex!=null){
                //异常处理逻辑
            }else {
                //正常处理逻辑
            }*/
        },SCHEDULER_POOL);
        return channelFuture;
    }
    /**
     * 发送响应包
     * @param responseMessage
     * @return ChannelFuture 如果有其他独特的处理监听处理可以自己搞，该方法默认只记录一个日志
     */
    public ChannelFuture sendResponse(CustomMessage responseMessage){
        ChannelFuture channelFuture = this.channel.writeAndFlush(responseMessage).addListener(f -> {
            if (!f.isSuccess()) {
                log.error("发送响应消息出错，消息体的json化字符串为{},具体错误为{}", JSON.toJSONString(responseMessage),f.cause());
            }
        });
        return channelFuture;
    }

}
